package com.zjw.server.socket;

import com.alibaba.cloud.nacos.discovery.NacosDiscoveryClient;
import com.zjw.model.R;
import com.zjw.model.constant.Constants;
import com.zjw.model.enums.ClientType;
import com.zjw.model.enums.NotifyType;
import com.zjw.model.pojo.vo.WsNotifyVO;
import com.zjw.server.exception.WebSocketException;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.commons.util.InetUtils;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.web.client.RestTemplate;

import javax.annotation.Resource;
import javax.websocket.Session;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * session容器
 *
 * @author zjw
 */
@Component
public class SessionContainer {

    private final static Logger log = LoggerFactory.getLogger(SessionContainer.class);

    @Resource
    private RestTemplate restTemplate;
    @Resource
    private InetUtils inetUtils;
    @Resource
    private NacosDiscoveryClient nacosDiscoveryClient;

    @Value("${ws.register.serviceName}")
    private String serviceName;

    @Value("${server.port}")
    private Integer port;


    /**
     * 存放本机所有在线的客户端：分组id，用户id，用户session
     */
    private Map<String, Map<String, Map<ClientType, Session>>> sessionContainer = new ConcurrentHashMap<>();

    /**
     * 失败队列
     */
    private ConcurrentLinkedQueue<WsNotifyVO> notifyQueue = new ConcurrentLinkedQueue<>();

    public void addSession(String groupId, String userId, ClientType clientType, Session session) {
        Map<ClientType, Session> user = getUserAllSession(groupId, userId);
        Session oldSession = user.get(clientType);
        if (oldSession != null) {
            try {
                session.close();
            } catch (IOException e) {
                log.error("close session error", e);
            }
        }
        user.put(clientType, session);
        //通知register
        WsNotifyVO notifyVO = new WsNotifyVO();
        notifyVO.setGroupId(groupId);
        notifyVO.setUserId(userId);
        notifyVO.setClientType(clientType);
        notifyVO.setNotifyType(NotifyType.ONLINE);
        notifyRegister(notifyVO);
    }

    public void removeSession(String groupId, String userId, ClientType clientType) {
        Map<String, Map<ClientType, Session>> group = sessionContainer.get(groupId);
        if (group == null) {
            return;
        }
        Map<ClientType, Session> user = group.get(userId);
        if (user == null) {
            return;
        }
        user.remove(clientType);
        //通知register
        WsNotifyVO notifyVO = new WsNotifyVO();
        notifyVO.setGroupId(groupId);
        notifyVO.setUserId(userId);
        notifyVO.setClientType(clientType);
        notifyVO.setNotifyType(NotifyType.OFFLINE);
        notifyRegister(notifyVO);
    }

    public Map<ClientType, Session> getUserAllSession(String groupId, String userId) {
        sessionContainer.putIfAbsent(groupId, new ConcurrentHashMap<>());
        Map<String, Map<ClientType, Session>> group = sessionContainer.get(groupId);
        group.putIfAbsent(userId, new ConcurrentHashMap<>());
        return group.get(userId);
    }

    public Session getSession(String groupId, String userId, ClientType clientType) {
        Map<ClientType, Session> user = getUserAllSession(groupId, userId);
        Session session = user.get(clientType);
        if (session == null) {
            throw new WebSocketException("session is empty");
        }
        return session;
    }

    /**
     * 发送消息:同步
     *
     * @param session
     * @param message
     * @return 消息是否发送成功
     */
    public boolean sendMessage(Session session, String message) {
        if (StringUtils.isBlank(message)) {
            throw new WebSocketException("message is empty");
        }
        try {
            session.getBasicRemote().sendText(message);
            return true;
        } catch (Exception e) {
            log.error("sendMessage error", e);
        }
        return false;
    }

    /**
     * 通知register
     *
     * @param notifyVO
     */
    public void notifyRegister(WsNotifyVO notifyVO) {
        for (int i = 0; i < 2; i++) {
            R<Void> result = notify(notifyVO);
            if (result.getCode() == Constants.SUCCESS) {
                return;
            }
            log.error("notifyRegister error,param={} result={}", notifyVO, result);
        }
        //加入失败队列等待重试
        notifyQueue.add(notifyVO);
    }

    /**
     * 重新通知register
     */
    public void retryNotifyRegister() {
        WsNotifyVO oldNotifyVO = null;
        while (!notifyQueue.isEmpty()) {
            WsNotifyVO notifyVO = notifyQueue.peek();
            if (notifyVO.equals(oldNotifyVO)) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    log.info("retryNotifyRegister sleep InterruptedException", e);
                }
            }
            oldNotifyVO = notifyVO;
            R<Void> result = notify(notifyVO);
            if (result.getCode() == Constants.SUCCESS) {
                notifyQueue.remove(notifyVO);
                break;
            }
            log.error("retryNotifyRegister error,param={} result={}", notifyVO, result);
        }
    }

    private R<Void> notify(WsNotifyVO notifyVO) {
        boolean flag = true;
        notifyVO.setPort(port);
        notifyVO.setIp(inetUtils.findFirstNonLoopbackHostInfo().getIpAddress());
        List<ServiceInstance> instances = nacosDiscoveryClient.getInstances(serviceName);
        if(CollectionUtils.isEmpty(instances)){
            log.error("notify error: register service is empty");
            return R.fail("register service is empty");
        }
        for (ServiceInstance instance : instances) {
            try {
                R<Void> result = restTemplate.postForObject(instance.getUri() + "/ws/register", notifyVO, R.class);
                if (result != null && result.getCode() == Constants.SUCCESS) {
                    continue;
                }
            } catch (Exception e) {
                log.error("notify error", e);
            }
            flag = false;
        }
        return flag ? R.ok() : R.fail();
    }
}
